package com.example.mq.mqserver.datacenter;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: ws
 * Date: 2023-10-03
 * Time: 17:00
 */
/*
 * 使用这个类来统一管理内存中的所有数据
 * 该类后续提供的一些方法，可能会在多线程环境下被使用，因此要注意线程安全
 */
public class MemoryDataCenter {
    // key 是 exchangeName，value 是 Exchange对象，用于查找交换机
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key 是 queueName，value 是 MSGQueue对象，用于查找消息队列
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 第一个 key 是 exchangeName，第二个 key 是 queueName，用于查找交换机绑定的所有消息队列
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // key 是 messageId，value 是 Message对象，用于查找消息
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // key 是 queueName，value 是一个 Message链表，用于查找队列存储的的所有消息
    // TODO List 可能不是线程安全，需要换成线程安全的
    private ConcurrentHashMap<String, List<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 第一个 key 是 queueName，第二个 key 是 messageId，用于查找队列存储的的”未被确认“的消息
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();



    /*
     * 针对交换机的操作
     */
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功！exchangeName=" + exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功！exchangeName=" + exchangeName);
    }

    /*
     * 针对队列的操作
     */
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 队列添加成功！queueName=" + queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功！queueName=" + queueName);
    }

    /*
     * 针对绑定的操作
     * 要保证线程安全
     */
    public void insertBinding(Binding binding) throws MqException {
        // 先使用 exchangeName 查一下，看对应的哈希表是否存在，不存在就创建
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
//        if (bindingMap == null) {
//            bindingMap = new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchangeName(), bindingMap);
//        }
        // 等价写法 computeIfAbsent
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
        synchronized (bindingMap) {// 保证判断和插入的线程安全
            // 再根据 queueName 查一下，如果已经已经存在就抛出异常，不存在插入
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter 绑定已经存在！ exchangeName=" + binding.getExchangeName() + "，queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
            System.out.println("[MemoryDataCenter] 绑定添加成功！exchangeName=" + binding.getExchangeName() + "，queueName=" + binding.getQueueName());
        }
    }

    /*
     * 获取绑定，写两个版本
     * 1. getBinding 根据 exchangeName 和 queueName 确定唯一一个 Binding
     * 2. getBindings 根据 exchangeName 获取到所有的 Binding
     *
     */
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            throw new MqException("[MemoryDataCenter] 绑定不存在！exchangeName=" + binding.getExchangeName() + "，queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功！exchangeName=" + binding.getExchangeName() + "，queueName=" + binding.getQueueName());
    }

    /*
     * 针对消息的操作
     */
    // 添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 新消息添加成功！messageId=" + message.getMessageId());
    }

    // 根据 id 查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    // 根据 id 删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除成功！messageId=" + messageId);
    }


     // 发送消息到指定的队列
     public void sendMessage(MSGQueue queue, Message message) {
        // 把消息队列放到对应的数据结构中
         // 根据队列的名字找到对应的消息链表
//         List<Message> messages = queueMessageMap.get(queue.getName());
//         if (messages == null) {
//             messages = new LinkedList<>();
//             queueMessageMap.put(queue.getName(), messages);
//         }
         // 替换
         List<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
         // 把数据加到链表中
         synchronized (messages) {
             messages.add(message);
         }
         // 把消息加入到消息中心，即使重复插入也无所谓
         addMessage(message);
         System.out.println("[MemoryDataCenter] 消息被投递到队列中！messageId=" + message.getMessageId());
     }

    // 从队列中取消息
    public Message pollMessage(String queueName) {
        // 根据队列名，查找对应的链表
        List<Message> messages = queueMessageMap.get(queueName);
        // 没找到
        if (messages == null) {
            return null;
        }
        synchronized (messages) {
            if (messages.size() == 0) {
                return null;
            }
            // 有元素，就进行头删
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出！messageId=" + currentMessage.getMessageId());
            return currentMessage;
        }
    }

    // 获得指定队列中的消息个数
    public int getMessageCount(String queueName) {
        List<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    // 添加未确认消息
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列！messageId=" + message.getMessageId());
    }

    // 删除已经确认过的消息
    public void removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除！messageId=" + messageId);
    }

    // 获取指定的未确认消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            return null;
        }
        return messageHashMap.get(messageId);
    }

    // 从硬盘上读取数据，把硬盘上持久化的数据恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        // 0. 清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 1. 恢复所有的队列数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        // 2. 恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }
        // 3. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        // 4. 恢复所有的消息数据
        // 遍历所有的队列，根据每个队列的名字，获取所有的消息
        for (MSGQueue queue : queues) {
            List<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }
        // 注意！！针对”未确认的消息“这部分内存中的数据，不要需要从硬盘恢复，之前考虑硬盘存储的时候，也没设定这一块
        // 一旦在等待 ack 的过程中，服务器重启了，此时这些”未被确认的消息“，就恢复成了”未被取走的消息“
        // 这个消息在硬盘上存储的时候，就当作”未被取走“，消费者需要再重新取
    }
}
